package cn.com.sparkle.paxos.client;

import java.io.IOException;
import java.io.Serializable;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.google.protobuf.ByteString;

import cn.com.sparkle.paxos.client.PaxosClient.CommandCallBack;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.net.NetNode;
//import cn.com.sparkle.paxos.client.net.message.AddRequest;
//import cn.com.sparkle.paxos.net.message.AddResponse;
//import cn.com.sparkle.paxos.net.message.MessagePackage;
import cn.com.sparkle.raptor.core.transport.socket.nio.exception.SessionHavaClosedException;

public class PaxosOperater {
	private final static Logger logger = Logger.getLogger(PaxosOperater.class);

	private final static int DEFULT_ADD_TIMEOUT = 5000;

	private static class AddCallable implements Callable<AddResponse> {
		public volatile AddResponse result;

		@Override
		public AddResponse call() throws Exception {
			return result;
		}
	}

	public static interface CallBack {
		public void callBack(AddResponse response);
	}
	public static interface WatchCallBack{
		public void watch(byte[] bytes);
	}

	private class ASyncCommandCallBack implements CommandCallBack {
		private AddCallable addCallable;
		private FutureTask<AddResponse> future;
		private CallBack customCallback;

		public ASyncCommandCallBack(AddCallable addCallable,
				FutureTask<AddResponse> future, CallBack customCallback) {
			super();
			this.addCallable = addCallable;
			this.future = future;
			this.customCallback = customCallback;
		}

		@Override
		public void response(NetNode node, MessagePackage response) {
			addCallable.result = response.getAddResponse();
			future.run();
			try {
				waitLock.lock();
				int r = waitFinishCounter.decrementAndGet();
				// logger.debug("wait finish add command size:" + r);
				if (isWaitFinish && waitFinishCounter.get() == 0) {
					allDoCondition.signal();
					isWaitFinish = false;
				}
				if (this.customCallback != null) {

					try {
						this.customCallback.callBack(response.getAddResponse());
					} catch (com.google.protobuf.UninitializedMessageException e) {
						try {
							System.out.println(" result---------------"
									+ new String(response.getAddResponse()
											.getResult().toByteArray())
									+ " "
									+ response.hasAddResponse()
									+ response.hasMasterBeatHeartRequest()
									+ "  "
									+ response.getId()
									+ " "
									+ node.getSession().getChannel()
											.getRemoteAddress().toString());
						} catch (IOException e1) {
							// TODO Auto-generated catch block
							e1.printStackTrace();
						}
						throw e;
					}
				}
			} finally {
				waitLock.unlock();
			}

		}
	}

	private CommandProcessor processor;

	public PaxosOperater(CommandProcessor processor) {
		super();
		this.processor = processor;
	}

	private AtomicInteger waitFinishCounter = new AtomicInteger(0);
	private ReentrantLock waitLock = new ReentrantLock();
	private Condition allDoCondition = waitLock.newCondition();
	private boolean isWaitFinish = false;

	public void waitAllFinish(long timeout) throws InterruptedException,
			MasterMayBeLostException {
		try {
			waitLock.lock();
			isWaitFinish = true;
			if (waitFinishCounter.get() != 0) {
				allDoCondition.await(timeout, TimeUnit.MILLISECONDS);
			}
			isWaitFinish = waitFinishCounter.get() != 0;
			if (isWaitFinish) {
				throw new MasterMayBeLostException();
			}

		} finally {
			waitLock.unlock();
		}
	}

	/**
	 * This is an async method.
	 * 
	 * @throws SessionHavaClosedException
	 * @throws InterruptedException
	 * @throws MasterMayBeLostException 
	 */
	public Future<AddResponse> add(byte[] value, boolean isTransportMaster)
			throws InterruptedException, MasterMayBeLostException {
		return add(value, DEFULT_ADD_TIMEOUT, isTransportMaster, null);
	}
	public Future<AddResponse> add(byte[] value, int timeout,
			boolean isTransportMaster, CallBack customCallback)
			throws InterruptedException, MasterMayBeLostException {
		AddCallable callable = new AddCallable();
		FutureTask<AddResponse> future = new FutureTask<AddResponse>(callable);
		// AddRequest request = new AddRequest(value);
		CommandCallBack commandCallBack = new ASyncCommandCallBack(callable,
				future, customCallback);
		waitFinishCounter.addAndGet(1);
		AddRequest.Builder builder = AddRequest.newBuilder()
				.setIsManagerCommand(false)
				.setTransportMaster(isTransportMaster)
				.setValue(ByteString.copyFrom(value));
		// MessagePackage.Builder mp =
		// MessagePackage.newBuilder().setId(-1).setIsLast(true).setAddRequest(builder);

		if (!processor.addCommand(
				new Command(builder.build(), commandCallBack), timeout,
				TimeUnit.MILLISECONDS)) {
			waitFinishCounter.decrementAndGet();
			throw new MasterMayBeLostException();
		}
		return future;
	}

	/**
	 * This is an sync method
	 * 
	 * @param value
	 * @throws InterruptedException
	 * @throws TimeoutException
	 * @throws MasterMayBeLostException 
	 * @throws ExecutionException
	 * @throws SessionHavaClosedException
	 */
	public AddResponse syncAdd(byte[] value, boolean isTransportMaster)
			throws InterruptedException, TimeoutException, MasterMayBeLostException {
		Future<AddResponse> f = add(value, isTransportMaster);
		try {
			return f.get(DEFULT_ADD_TIMEOUT, TimeUnit.MILLISECONDS);
		} catch (ExecutionException e) {
			throw new RuntimeException("fatal exception", e);
		}
	}
}
